#include "OpcListener.hpp"
#include "ThreadPool.hpp"

OpcListener::OpcListener()
{
	m_client = UA_Client_new();
	UA_ClientConfig*config = UA_Client_getConfig(m_client);
	config->timeout = DefaultTimeout;
	UA_ClientConfig_setDefault(config);
}

OpcListener::~OpcListener()
{
	stop();
	if(m_client)
	{
		UA_Client_delete(m_client);
	}
}

bool OpcListener::connect(const std::string & endpointUrl)
{
	if(m_running)
	{
		std::cout << "OpcListener is already running" << std::endl;
		return false;
	}
	m_endpointUrl = endpointUrl;
	m_running = true;
	if(m_connected = internalConnect())
	{
		std::cout << "OpcListener connected[" << m_endpointUrl << "]" << std::endl;
	}
	m_workerThread = std::thread(&OpcListener::worker, this);
	return m_connected;
}

void OpcListener::disconnect()
{
	m_connected = false;
	if(m_client == nullptr)
	{
		return;
	}
	if(m_subscriptionId != 0)
	{
		UA_Client_Subscriptions_deleteSingle(m_client, m_subscriptionId);
		m_subscriptionId = 0;
	}
	UA_Client_disconnect(m_client);
	// clear listener items
	std::cout << "OpcListener disconnected" << std::endl;
}

bool OpcListener::isConnected() const
{
	return m_connected;
}

bool OpcListener::isRunning() const
{
	return m_running;
}

bool OpcListener::addListener(const UA_NodeId & nodeId, OpcListenerHandler handler)
{
	UA_MonitoredItemCreateRequest itemRequest = UA_MonitoredItemCreateRequest_default(nodeId);
	itemRequest.requestedParameters.samplingInterval = m_intervalMs;
	itemRequest.monitoringMode = UA_MONITORINGMODE_REPORTING;
	//itemRequest.requestedParameters.queueSize = 10;
	itemRequest.requestedParameters.discardOldest = true;

	UA_Client_DataChangeNotificationCallback callback = DataChangeCallback;

	for(int i = 0; i < MaxRetryCount; ++i)
	{
		//itemRequest.requestedParameters.clientHandle = ++m_counter; //m_subscriptionId;//++m_counter;
		UA_MonitoredItemCreateResult result = UA_Client_MonitoredItems_createDataChange(
		                                          m_client, m_subscriptionId, UA_TIMESTAMPSTORETURN_BOTH,
		                                          itemRequest, this, callback, nullptr
		                                      );

		if(result.statusCode != UA_STATUSCODE_GOOD)
		{
			std::cout << "Failed to create monitored item for NodeId ns=" << nodeId.namespaceIndex << "s=" << nodeId.identifier.string.data << std::endl;
			std::cout << RetryInterval << "ms later, retry[" << i + 1 << "/" << MaxRetryCount << "]" << std::endl;
			std::this_thread::sleep_for(std::chrono::milliseconds(RetryInterval));
			continue;
		}

		// 存储监视项信息
		{
			std::lock_guard<std::mutex> lock(m_itemMutex);
			if(m_listenerItems.find(result.monitoredItemId) != m_listenerItems.end())
			{
				std::cout << "Monitored item[" << nodeId.identifier.string.data << "] for NodeId already exists." << std::endl;
			}
			UA_Variant variant;
			UA_Variant_init(&variant);
			OpcListenerItem item = {std::move(variant), nodeId, handler, result.monitoredItemId, std::string((char*)nodeId.identifier.string.data, nodeId.identifier.string.length)};
			std::memcpy(&item.nodeId, &nodeId, sizeof(UA_NodeId));
			item.nodeId.identifier.string = UA_STRING_ALLOC(item.nodeName.c_str());
			m_listenerItems.insert({result.monitoredItemId, std::move(item)});
		}
		std::cout << "Monitored item created for NodeId ns=" << nodeId.namespaceIndex << "s=" << nodeId.identifier.string.data << std::endl;
		return true;
	}
	return false;
}

bool OpcListener::removeListener(const UA_NodeId & nodeId)
{
	UA_UInt32 itemIdToRemove = 0;
	{
		std::lock_guard<std::mutex> lock(m_itemMutex);

		for(auto it = m_listenerItems.begin(); it != m_listenerItems.end(); ++it)
		{
			if(UA_NodeId_equal(&it->second.nodeId, &nodeId))
			{
				itemIdToRemove = it->second.monitoredItemId;
				m_listenerItems.erase(itemIdToRemove);
				break;
			}
		}
	}

	if(itemIdToRemove == 0)
	{
		std::cout << "Monitored item for NodeId not found." << std::endl;
		return false;
	}

	UA_StatusCode retval = UA_Client_MonitoredItems_deleteSingle(m_client, m_subscriptionId, itemIdToRemove);
	if(retval != UA_STATUSCODE_GOOD)
	{
		std::cerr << "Failed to delete monitored item: " << UA_StatusCode_name(retval) << std::endl;
		return false;
	}

	std::cout << "Monitored item removed for NodeId." << std::endl;
	return true;
}

void OpcListener::stop()
{
	m_running = false;
	if(m_workerThread.joinable())
	{
		m_workerThread.join();
	}

	disconnect(); // 确保断开连接

	// 通知可能正在等待的线程
	{
		std::lock_guard<std::mutex> lock(m_cvMutex);
		m_cv.notify_all();
	}
	for(auto& item : m_listenerItems)
	{
		UA_Variant_clear(&item.second.value);
		UA_NodeId_clear(&item.second.nodeId);
	}
	m_listenerItems.clear();
	m_counter = 0;

	std::cout << "OpcListener stopped" << std::endl;
}

void OpcListener::setIntervalMs(int intervalMs)
{
	m_intervalMs = intervalMs;
}

void OpcListener::setReconnectParameters(int intervalMs, int maxAttempts)
{
	m_reconnectIntervalMs = intervalMs;
	m_maxReconnectAttempts = maxAttempts;
}

void OpcListener::DataChangeCallback(UA_Client* client, UA_UInt32 subId, void* subContext, UA_UInt32 monId, void *monContext, UA_DataValue* value)
{
	OpcListener* instance = static_cast<OpcListener *>(subContext);
	if(!instance || !value->hasValue)
	{
		return;
	}

	OpcListenerHandler handler = nullptr;
	{
		std::lock_guard<std::mutex> lock(instance->m_itemMutex);
		if(instance->m_listenerItems.find(monId) != instance->m_listenerItems.end())
		{
			handler = instance->m_listenerItems[monId].handler;
			UA_Variant_copy(&value->value, &instance->m_listenerItems[monId].value);
		}
	}

	if(handler)
	{
		gT.push([ = ]()
		{
			std::lock_guard<std::mutex> lock(instance->m_itemMutex);
			handler(instance->m_listenerItems[monId].nodeId, instance->m_listenerItems[monId].value);
		});
	}
}

bool OpcListener::internalConnect()
{
	UA_StatusCode retval = UA_Client_connect(m_client, (char*)m_endpointUrl.c_str());
	if(retval != UA_STATUSCODE_GOOD)
	{
		std::cerr << "Failed to connect to OPC UA server: " << UA_StatusCode_name(retval) << std::endl;
		return false;
	}

	UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
	request.requestedPublishingInterval = m_intervalMs;//数据发布间隔?
	request.requestedLifetimeCount = 1000; //订阅的生命周期计数器? 次数*间隔
	request.requestedMaxKeepAliveCount = 100; //控制服务器发送“无数据变更”通知的频率  次数*间隔
	request.maxNotificationsPerPublish = UA_UINT32_MAX;//单次发布通知中允许的最大消息数量
	request.publishingEnabled = true;//是否立即启用订阅?

	UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(m_client, request, this, nullptr, nullptr);
	if(response.responseHeader.serviceResult != UA_STATUSCODE_GOOD)
	{
		std::cerr << "Failed to create subscription: " << UA_StatusCode_name(response.responseHeader.serviceResult) << std::endl;
		UA_Client_disconnect(m_client);
		return false;
	}
	m_subscriptionId = response.subscriptionId;;
	std::cout << "Subscription created with ID: " << m_subscriptionId << std::endl;

	// 重新创建之前订阅的监视项
	std::lock_guard<std::mutex> lock(m_itemMutex);
	auto itemsToRecreate = m_listenerItems; // 复制一份
	m_listenerItems.clear(); // 清空旧的，因为ID会变
	for(auto& item : itemsToRecreate)
	{
		// 重置旧ID
		item.second.monitoredItemId = 0;
		UA_MonitoredItemCreateRequest itemRequest = UA_MonitoredItemCreateRequest_default(item.second.nodeId);

		UA_Client_DataChangeNotificationCallback callback = DataChangeCallback;
		void *subContext = this;

		UA_MonitoredItemCreateResult result = UA_Client_MonitoredItems_createDataChange(
		                                          m_client, m_subscriptionId,
		                                          UA_TIMESTAMPSTORETURN_BOTH,
		                                          itemRequest, subContext, callback, nullptr
		                                      );

		if(result.statusCode != UA_STATUSCODE_GOOD)
		{
			std::cout << "Failed to re-create monitored item for NodeId ns=" << item.second.nodeId.namespaceIndex << "s=" << item.second.nodeId.identifier.string.data << ": " << UA_StatusCode_name(result.statusCode) << std::endl;
		}
		else
		{
			item.second.monitoredItemId = result.monitoredItemId;
			m_listenerItems.insert(item);
			std::cout << "Re-created monitored item for NodeId ns=" << item.second.nodeId.namespaceIndex << "s=" << item.second.nodeId.identifier.string.data << " with new ID: " << result.monitoredItemId << std::endl;
		}
	}

	return true;
}

void OpcListener::worker()
{
	while(isRunning())
	{
		if(isConnected())
		{
			updateBySubscription();
			std::this_thread::sleep_for(std::chrono::milliseconds(m_intervalMs));
		}
		else
		{
			// 未连接状态，尝试重连
			if(m_maxReconnectAttempts == ContinueRetry || m_currentReconnectAttempts < m_maxReconnectAttempts)
			{
				std::cout << "Attempting to reconnect..." << std::endl;

				if(internalConnect())
				{
					std::cout << "Reconnected successfully." << std::endl;
					m_connected = true;
					m_currentReconnectAttempts = 0; // 重置计数
				}
				else
				{
					m_currentReconnectAttempts++;
					if(isRunning())    // 再次检查running标志
					{
						std::cerr << "Reconnect failed. Waiting " << m_reconnectIntervalMs << "ms before retry..." << std::endl;
						// 等待重连间隔或被 stop 唤醒
						std::unique_lock<std::mutex> lock(m_cvMutex);
						m_cv.wait_for(lock, std::chrono::milliseconds(m_reconnectIntervalMs), [this] { return !m_running; });
					}
				}
			}
			else
			{
				std::cerr << "Maximum reconnect attempts reached. Stopping client." << std::endl;
				m_running = false; // 达到最大尝试次数，停止客户端
			}
		}
	}
	std::cout << "OpcListener Worker thread exited." << std::endl;
}

void OpcListener::updateBySubscription()
{
	// 处理网络事件和回调
	UA_StatusCode retval = UA_Client_run_iterate(m_client, 100);
	if(retval != UA_STATUSCODE_GOOD)
	{
		std::cerr << "OpcListener disconnected: " << UA_StatusCode_name(retval) << std::endl;
		m_connected = false;
		m_currentReconnectAttempts = 0; // 重置尝试次数

		// 在断开连接后立即清理客户端状态，为重连做准备
		UA_Client_disconnect(m_client);
		if(m_subscriptionId != 0)
		{
			// Subscription is gone, clear ID
			m_subscriptionId = 0;
		}
	}
}